package Examples;

import homomorphism.list.ThirdHomomorphismInterface;
import utility.LongPair;
import java.util.ArrayList;
import java.util.Iterator;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.io.LongWritable;
import utility.PairT;

/**
 *
 * @author takeshi
 *  for test
 */
public class userTest implements ThirdHomomorphismInterface<LongWritable, LongPair> {

    private static final Log LOG = LogFactory.getLog(userTest.class);

    public static LongWritable mps(ArrayList<LongWritable> xs) {
        Long rst = 0l;
        Long tmp = 0l;
        for (LongWritable v : xs) {
            tmp += v.get();
            if (tmp > rst) {
                rst = tmp;
            }
        }
        return new LongWritable(rst);
    }

    public static PairT<Long> mps_sum(ArrayList<LongWritable> xs) {
        Long mps = 0l;
        Long sum = 0l;
        int count = 0;
        for (LongWritable v : xs) {
            sum += v.get();
            if (sum > mps) {
                mps = sum;
            }
            LOG.info("mps_sum " + count + " v  is" + v.get() + " mps is" + mps + "  sum  is " + sum);
            count++;
        }
        LOG.info("xs length is " + xs.size() + "mps_sum = (" + new PairT<Long>(mps, sum) + ")");
        return new PairT<Long>(mps, sum);


    }

    public static PairT<Long> mps_sum(Iterator<LongWritable> xs) {
        Long mps = 0l;
        Long sum = 0l;
        while (xs.hasNext()) {
            sum += xs.next().get();
            if (sum > mps) {
                mps = sum;
            }
        }
        return new PairT<Long>(mps, sum);
    }

    public static LongWritable sum(ArrayList<LongWritable> xs) {
        Long rst = 0l;
        for (LongWritable var : xs) {
            rst += var.get();
        }
        return new LongWritable(rst);
    }

    public LongPair fold(ArrayList<LongWritable> values) {
        PairT<Long> mps_sum = mps_sum(values);
        LongPair rst = new LongPair(mps_sum);
        //LongWritable mps = mps(values);
        //LongWritable sum = sum(values);
        // rst.Set(mps.get(), sum.get());
        return rst;
    }

    public LongPair fold(Iterator<LongWritable> values) {
        PairT<Long> mps_sum = mps_sum(values);
        LongPair rst = new LongPair(mps_sum);
        //LongWritable mps = mps(values);
        //LongWritable sum = sum(values);
        // rst.Set(mps.get(), sum.get());
        return rst;
    }

    public LongPair fold(Iterable<LongWritable> values) {
        PairT<Long> mps_sum = mps_sum(values);
        LongPair rst = new LongPair(mps_sum);
        return rst;
    }

    public ArrayList<LongWritable> unfold(LongPair value) {
        ArrayList<LongWritable> rst = new ArrayList<LongWritable>();
        rst.clear();
        long tmp = value.getFirst();
        rst.add(new LongWritable(tmp));
        long tmp2 = (value.getSecond() - tmp);
        rst.add(new LongWritable(tmp2));
        return rst;
    }

    public static PairT<Long> mps_sum(Iterable<LongWritable> xs) {
        Long mps = 0l;
        Long sum = 0l;
        for (LongWritable lw : xs) {
            sum += lw.get();
            if (sum > mps) {
                mps = sum;
            }
        }
        return new PairT<Long>(mps, sum);
    }
    /**
     * The main method.
     *
     * usage:  [input file path]  [output file path] (option [job name 1] [job name 2]) -h  :[usage]
     *
     */
//    public static void main(String[] args) throws Exception {
//
//        int exitCode = LHRunner.InvokeRun(new userTest(), args);
//        System.exit(exitCode);
//
//    }
//        public static void main(String[] args) throws Exception {
//
//        if (args.length < 2) {
//            //   System.out.println("the args.length = " + args.length);
//            for (int i = 0; i < args.length; i++) {
//                System.err.println(args[i]);
//            }
//            System.err.println("Usage:hadoop jar  <input path> <output path> <userClass>,...");
//            System.exit(-1);
//        }
//      //  Class userC = Class.forName(args[2].trim()+".class");
//       // Class foldingType = null;
//        if(args.length > 3){
//         //   foldingType = Class.forName(args[3].trim()) ;
//        }
//        int res = ToolRunner.run(new JobConf(), new userTest(), args);
//        System.exit(res);
//
//    }
}
